package com.atguigu.flink.exec2;

import com.atguigu.flink.function.MyUtil;
import com.atguigu.flink.pojo.UserBehavior;
import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
import org.apache.flink.api.common.RuntimeExecutionMode;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.AggregateFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.state.ListState;
import org.apache.flink.api.common.state.ListStateDescriptor;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.TimerService;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.SlidingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.util.List;
import java.util.stream.Collectors;
import java.util.stream.StreamSupport;

/**
 * Created by Smexy on 2022/12/17
 *
 * 每隔5分钟输出最近1小时  内  点击量最多的前N个商品
 *
 * 数据粒度： 一行是一个用户对一种商品点击一次
 *
 *  开时间窗口
 *       slide = 5min
 *       size  = 1h
 *
 *  -------------------------------
 *    点击量最多  的  前N个  商品
 *
 *      第一步： 统计每种商品被点击的总次数
 *          按照itemId分组，开窗1h,sum统计
 *           [8点,9点)   a,100
 *           [8点,9点)   b,200
 *           [9点,10点)  c,300,
 *           [9点,10点)  d,400
 *
 *      第二步: 将同一窗口中统计的结果，再求前3
 *            按照窗口分组，再统计前3
 *
 *            什么时候执行?
 *              在所计算的窗口的第一条数据到达后，注册一个和窗口的endtime一致的定时器。
 *
 *              到点就执行。
 *
 *              当下游的算子的时间被更新为 2.00时，说明上游的每一个subTask 2:00之前的数据都已经全部发送到下游了，可以开始进行排序了！
 *
 */
public class Demo3_HotItem
{
    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();


        WatermarkStrategy<UserBehavior> watermarkStrategy = WatermarkStrategy.<UserBehavior>forMonotonousTimestamps()
            .withTimestampAssigner( (e, r) -> e.getTimestamp() * 1000);

        env
            .readTextFile("data/UserBehavior.csv")
            .map(new MapFunction<String, UserBehavior>()
            {
                @Override
                public UserBehavior map(String value) throws Exception {
                    String[] words = value.split(",");
                    UserBehavior userBehavior = new UserBehavior(
                        Long.valueOf(words[0]),
                        Long.valueOf(words[1]),
                        Integer.valueOf(words[2]),
                        words[3],
                        Long.valueOf(words[4])
                    );
                    return userBehavior;
                }
            })
            .assignTimestampsAndWatermarks(watermarkStrategy)
            .filter(u -> "pv".equals(u.getBehavior()))
            .keyBy(UserBehavior::getItemId)
            .window(SlidingEventTimeWindows.of(Time.hours(1),Time.minutes(5)))
            .aggregate(new AggregateFunction<UserBehavior, Integer, HotItem>()
            {
                @Override
                public Integer createAccumulator() {
                    return 0;
                }

                @Override
                public Integer add(UserBehavior value, Integer accumulator) {
                    return accumulator += 1;
                }

                @Override
                public HotItem getResult(Integer accumulator) {
                    return new HotItem(null, null, null, accumulator);
                }

                @Override
                public Integer merge(Integer a, Integer b) {
                    return null;
                }
            }, new ProcessWindowFunction<HotItem, HotItem, Long, TimeWindow>()
            {
                //开窗中的process，仅仅执行一次
                @Override
                public void process(Long key, Context context, Iterable<HotItem> elements, Collector<HotItem> out) throws Exception {
                    //取出聚合的结果
                    HotItem hotItem = elements.iterator().next();

                    hotItem.setItemId(key);
                    TimeWindow window = context.window();
                    //System.out.println(MyUtil.printTimeWindow(window));

                    hotItem.setStart(window.getStart());
                    hotItem.setEnd(window.getEnd());

                    out.collect(hotItem);

                }
            })
            .keyBy(HotItem::getEnd)
            //不开窗的，每一条数据执行一次
            .process(new KeyedProcessFunction<Long, HotItem, String>()
            {

                private ListState<HotItem> items;
                //判断是否时窗口中的第一条的标记
                private ValueState<Boolean> isFirst;

                //完成top3排序
                @Override
                public void onTimer(long timestamp, OnTimerContext ctx, Collector<String> out) throws Exception {

                    List<HotItem> top3 = StreamSupport.stream(items.get().spliterator(), true)
                                                         .sorted((i1, i2) -> -i1.getClick().compareTo(i2.getClick()))
                                                         .limit(3)
                                                         .collect(Collectors.toList());

                    HotItem hotItem = top3.get(0);

                    String windowStr = MyUtil.printTimeWindow2(new TimeWindow(hotItem.getStart(), hotItem.getEnd()));

                    String top3Str = top3.stream().map(i -> i.getItemId() + "->" + i.getClick()).collect(Collectors.joining(","));

                    out.collect(windowStr + ":" + top3Str);

                }

                @Override
                public void open(Configuration parameters) throws Exception {
                    isFirst = getRuntimeContext().getState(new ValueStateDescriptor<>("isFirst", Boolean.class));
                    items = getRuntimeContext().getListState(new ListStateDescriptor<>("items", HotItem.class));
                }

                //在某个窗口第一条数据到达时，注册定时器
                @Override
                public void processElement(HotItem value, Context ctx, Collector<String> out) throws Exception {

                    TimerService timerService = ctx.timerService();

                    //如果时第一条 isFirst里面保存的变量是没有被赋值的
                    if (isFirst.value() == null){

                        //注册定时器
                        timerService.registerEventTimeTimer(value.getEnd() + 5000);
                        //更新标记
                        isFirst.update(false);

                    }

                    //把数据加入到集合中
                    items.add(value);
                }
            })//.setParallelism(1)
            .print().setParallelism(1);

        env.execute();
    }

    @NoArgsConstructor
    @AllArgsConstructor
    @Data
    private static class HotItem {

        //标识窗口的时间范围
        private Long start;
        private Long end;
        private  Long itemId;
        private  Integer click;

    }
}
